package com.eshipenet.shipnet.edgechannelbuilder

import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema

class EdgeChannelActionKafkaSerializer
    extends KeyedSerializationSchema[EdgeChannelAction] {

  override def serializeKey(element: EdgeChannelAction): Array[Byte] =
    element match {
      case EdgeChannelInsert(after)    => after.id.getBytes()
      case EdgeChannelUpdate(_, after) => after.id.getBytes()
      case EdgeChannelDelete(before)   => before.id.getBytes()
    }

  override def serializeValue(element: EdgeChannelAction): Array[Byte] =
    element.toJson.toString().getBytes()

  override def getTargetTopic(element: EdgeChannelAction): String =
    "EdgeChannel"

}
